/*
 * File:    Worker.java
 * Created: 04-Jun-2007
 * Version: $Id$
 * 
 * COPYRIGHT (C) 2007, Bitgate Software, LLC.  All Rights Reserved.
 * RELEASED UNDER THE CREATIVE COMMONS LICENSE VERSION 2.5.
 *
 * $Log$
 * 
 * software@bitgatesoftware.com
 */

package com.bitgate.util.service;

import static com.bitgate.util.debug.Debug.debug;
import static com.bitgate.util.debug.Debug.getStackTrace;
import static com.bitgate.util.debug.Debug.info;
import static com.bitgate.util.debug.Debug.isDebugEnabled;

import java.io.IOException;
import java.lang.ref.WeakReference;
import java.nio.CharBuffer;
import java.nio.charset.Charset;
import java.nio.charset.CharsetDecoder;
import java.util.concurrent.atomic.AtomicInteger;

import com.bitgate.util.service.client.ClientContext;
import com.bitgate.util.service.protocol.ProtocolInterface;
import com.bitgate.util.shutdown.ShutdownHook;
import com.bitgate.util.shutdown.ShutdownHookInterface;
import com.bitgate.util.socket.ClientChannel;
import com.bitgate.util.threads.DynamicThread;

/**
 * This is the worker object that handles connections from the underlying <code>Service</code> object, and handles its requests.
 * The <code>Service</code> object ultimately fires off a certain number of <code>Worker</code> threads to handle incoming connections,
 * and rotates through available <code>Worker</code>s.
 * <p/>
 * The <code>Worker</code> is basically a transaction object that takes input from a <code>Service</code> and passes it through
 * to the underlying <code>ProtocolInterface</code> object, which does all the hard work.  The <code>ProtocolInterface</code>
 * handles the protocol layer of the transaction, so whatever is loaded into that class will retrieve the data to communicate
 * with the connected client.
 * <p/>
 * As a result, this code is very lightweight.
 * 
 * @author khollis
 */
public class Worker implements DynamicThread, ShutdownHookInterface
{
	private final WeakReference<ProtocolInterface> protocolInterface;
	private final WeakReference<Service> runningService;
	private final AtomicInteger activeClients = new AtomicInteger();
    private static final CharsetDecoder decoder = Charset.forName(Service.getSystemEncoding()).newDecoder();
    
    // Thread-safe non-synchronized use of the running flag; no need for atomic write/read here.
    private volatile boolean isRunning;
	
    /**
     * Constructs a new worker object.
     * 
     * @param protocolInterface The underlying <code>ProtocolInterface</code> that is attached to the <code>Service</code>.
     * @param runningService The <code>Service</code> object active for this connection.
     */
	public Worker(ProtocolInterface protocolInterface, Service runningService)
	{
		this.isRunning = true;
		this.protocolInterface = new WeakReference<ProtocolInterface>(protocolInterface);
		this.runningService = new WeakReference<Service>(runningService);
		
		if (protocolInterface == null) {
			throw new NullPointerException("protocolInterface");
		}
		
		if (runningService == null) {
			throw new NullPointerException("runningService");
		}
		
		ShutdownHook.getInstance().register(this);
	}
	
	/**
	 * Increases the number of active clients for this worker.
	 */
	private void increaseLoad()
	{
		int load = activeClients.incrementAndGet();
		
		if (isDebugEnabled()) {
			debug("Increased load, active clients = " + load);
		}
	}
	
	/**
	 * Decreases the number of active clients for this worker.
	 */
	private void decreaseLoad()
	{
		int load = activeClients.decrementAndGet();
		
		if (isDebugEnabled()) {
			debug("Decreased load, active clients = " + load);
		}
	}
	
	/**
	 * This is an entry-point class used by the <code>Service</code> object to tell the <code>Worker</code> that a
	 * request to handle a read or write operation.
	 * 
	 * @param commObject The <code>Object</code> containing the operation to perform.
	 */
	public void sendObject(Object commObject)
	{
		increaseLoad();
		
		try {
			if (commObject instanceof WorkerReadMessage) {
				handleRead(((WorkerReadMessage) commObject).getClientContext());
			} else if (commObject instanceof WorkerWriteMessage) {
				handleWrite(((WorkerWriteMessage) commObject).getClientContext());
			} else {
				info("Unknown object received: " + commObject.getClass().getName());
			}
		} catch(Exception e) {
			debug("Exception caught: " + e.getMessage() + ": " + e.getClass().getName());
		} finally {
			decreaseLoad();
		}
	}
    
	/**
	 * Handles the reading of data from the connected client.
	 * 
	 * @param cContext The <code>ClientContext</code> active for this session.
	 */
    private void handleRead(final ClientContext cContext)
    {
        int len = 0;
        
        if (cContext == null) {
        	throw new NullPointerException("ClientContext");
        }
        
        ClientChannel cChannel = cContext.getClientChannel();
        
        if (cChannel == null) {
        	throw new NullPointerException("ClientChannel");
        }

        try {
            len = cChannel.read(cContext.getByteBuffer());
        } catch(IOException e) {
        	if (isDebugEnabled()) {
        		debug("Unable to read from channel: " + e.getMessage());
        	}
        	
            protocolInterface.get().removeConnection(cContext);
            cContext.getByteBuffer().clear();
            return;
        }

        if (len == -1) {
        	if (isDebugEnabled()) {
        		debug("Client at serial '" + cContext.getSerialId() + "' disconnected.");
        	}
        	
            protocolInterface.get().removeConnection(cContext);

            try {
                cChannel.close();
            } catch(IOException e) {
            	if (isDebugEnabled()) {
            		debug("Unable to remove serial '" + cContext.getSerialId() + "' from client socket list.");
            	}
            }
        } else if (len > 0) {
        	cContext.getByteBuffer().flip();
            CharBuffer buf = null;
            
            try {
                buf = decoder.decode(cContext.getByteBuffer());
            } catch(Exception e) {
            	if (isDebugEnabled()) {
            		debug("Unable to decode input buffer: " + e);
            	}
            }

			try {
				if (buf != null) {
					if (isDebugEnabled()) {
						debug("Read data length=" + buf.length());
					}
					protocolInterface.get().processRead(cContext, buf.toString().getBytes());
				}
			} catch(Exception e) {
				if (isDebugEnabled()) {
					debug("Unable to process read via protocol interface: " + e.getClass().getName() + " caused by: \n" +
						getStackTrace(e).toString());
				}
			}
        }

        cContext.getByteBuffer().clear();
    }
    
    /**
     * Handles the write functionality for this client connection.
     * 
     * @param cContext The <code>ClientContext</code> for the currently active connection.
     */
    private void handleWrite(ClientContext cContext)
    {
    	if (!protocolInterface.get().processWrite(cContext)) {
    		if (cContext.getClientChannel().isOpen()) {
    			ServicePipe.registerRead(cContext, this.runningService.get().getConnectionPipe());
    		} else {
    			if (isDebugEnabled()) {
    				debug("Refrain from setting serial '" + cContext.getSerialId() + "', socket closed.");
    			}
    		}
    	} else {
    		if (isDebugEnabled()) {
    			debug("Continue writing.");
    		}
    	}
    }
    
    /**
     * Runs the worker.
     */
	public void run()
	{
		if (isDebugEnabled()) {
			debug("Worker started, waiting for connections from Service.");
		}
			
		while(isRunning) {
			try {
				synchronized(this) {
					wait(2000L);
				}
			} catch(Exception e) {
				if (isDebugEnabled()) {
					debug("Unable to sleep: Interrupted thread: " + e.getMessage());
				}
			}
		}
		
		info("Thread signalled to shutdown.  Shutting down at leisure.");
	}
	
	/**
	 * Creates a new instance of the <code>Worker</code> thread, which can be used by the <code>DynamicThreadGroup</code> object
	 * to create a new worker dynamically.
	 * 
	 * @return <code>DynamicThread</code> object.
	 */
	public DynamicThread newInstance()
	{
		try {
			return new Worker(protocolInterface.get(), runningService.get());
		} catch(Exception e) {
			if (isDebugEnabled()) {
				debug("Unable to create a new instance of this worker object.");
			}
			
			return null;
		}
	}
	
	/**
	 * Indicates whether or not this <code>Worker</code> is active.
	 * 
	 * @return <code>true</code> if the object is running, <code>false</code> otherwise.
	 */
	public boolean isRunning()
	{
		return isRunning;
	}
	
	/**
	 * Tells this thread to shutdown.
	 */
	public void shutdown()
	{
		isRunning = false;
	}
	
	/**
	 * Returns the current busy-state of the thread.
	 * 
	 * @return <code>DynamicThreadLoad</code> containing the thread state.
	 */
	public DynamicThreadLoad getBusyState()
	{
		if (activeClients.get() == 0) {
			return DynamicThreadLoad.IDLE;
		} else if (activeClients.get() >= 5) {
			return DynamicThreadLoad.HIGH;
		}
		
		return DynamicThreadLoad.NORMAL;
	}

	public boolean isShutdown() {
		return (!isRunning);
	}
}